package com.nl.pulsardemo.pool.config.consumer;


import lombok.extern.slf4j.Slf4j;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.pulsar.client.api.*;

import java.util.concurrent.TimeUnit;

import static com.nl.pulsardemo.pool.config.commons.Constants.SUBSCRIPTION_NAME;
import static com.nl.pulsardemo.pool.config.commons.Constants.TOPIC_NAME;
import static com.nl.pulsardemo.pool.config.commons.Constants.SERVICE_URL;

/**
 * @author : ZhaoY
 * @description :
 * @date : 2021-01-31 16:08
 **/
@Slf4j
public class PooledConsumerFactory implements PooledObjectFactory<Consumer> {
    private PulsarClient pulsarClient;

    public PooledConsumerFactory() {
//        try {
//            pulsarClient = PulsarClient.builder()
//                    .serviceUrl(SERVICE_URL)
//                    .build();
//        } catch (PulsarClientException e) {
//            log.info("PulsarClient初始化失败");
//            e.printStackTrace();
//        }
    }

    public PooledConsumerFactory(PulsarClient pulsarClient) {
        this.pulsarClient = pulsarClient;
    }

    @Override
    public PooledObject<Consumer> makeObject() throws Exception {
        return new DefaultPooledObject<Consumer>(pulsarClient.newConsumer(Schema.STRING)
                .topic(TOPIC_NAME)
                .subscriptionName(SUBSCRIPTION_NAME)
                .subscriptionType(SubscriptionType.Exclusive)//指定消费模式，包含：Exclusive，Failover，Shared，Key_Shared。默认Exclusive模式
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)//指定从哪里开始消费还有Latest，valueof可选，默认Latest
                .negativeAckRedeliveryDelay(60, TimeUnit.SECONDS)//指定消费失败后延迟多久broker重新发送消息给consumer，默认60s
                .subscribe());
    }

    @Override
    public void destroyObject(PooledObject<Consumer> pooledObject) throws Exception {
        pooledObject.getObject().close();

    }

    @Override
    public boolean validateObject(PooledObject<Consumer> pooledObject) {
        return false;
    }

    @Override
    public void activateObject(PooledObject<Consumer> pooledObject) throws Exception {

    }

    @Override
    public void passivateObject(PooledObject<Consumer> pooledObject) throws Exception {

    }
}
